package org.example.com.zwh.service;

import io.lettuce.core.RedisCommandTimeoutException;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Service
public class SeckillConsumer {
    private static final ExecutorService executor = Executors.newFixedThreadPool(5);

    // 新增健康监控依赖
    private final ConsumerHealthIndicator consumerHealthIndicator;

    // Redis操作模板（用于队列操作）
    private final RedisTemplate<String, Object> redisTemplate;
    // 秒杀业务处理服务
    private final OrderService orderService;
    private final SeckillService seckillService;

    /**
     * 构造器注入依赖
     * @param redisTemplate Redis操作模板
     * @param orderService 秒杀业务服务
     */
    public SeckillConsumer(RedisTemplate<String, Object> redisTemplate,
                           OrderService orderService,SeckillService seckillService,
                              ConsumerHealthIndicator consumerHealthIndicator) {
        this.redisTemplate = redisTemplate;
        this.orderService= orderService;
        this.seckillService= seckillService;
        this.consumerHealthIndicator = consumerHealthIndicator;

        }
    /**
     * 服务启动后初始化消费者线程
     * 使用@PostConstruct保证Spring Bean初始化完成后执行
     */
    @PostConstruct
    public void init() {
        // 启动消费者线程
        executor.submit(this::consumeTasks);
    }
    /**
     * 核心消费逻辑
     * 持续阻塞监听Redis队列，获取到任务后处理秒杀请求
     */
    private void consumeTasks() {
        System.out.println("消费线程已启动 ▶ " + Thread.currentThread().getName());

        consumerHealthIndicator.setConsumerRunning(true);

        while (true) {
            try {
                // 添加请求日志
                System.out.println("[" + new Date() + "] 等待获取队列任务...");
                Object task = redisTemplate.opsForList().rightPop("seckill:queue", 30, TimeUnit.SECONDS);

//                Object task = redisTemplate.opsForList().leftPop("seckill:queue", 30, TimeUnit.SECONDS);
                System.out.println("Received raw task: " + task); // 添加原始数据日志

//                Object task = redisTemplate.execute(new SessionCallback<Object>() {
//                    @Override
//                    public Object execute(RedisOperations operations) throws DataAccessException {
//                        // 阻塞式获取队列任务（30秒超时）
//                        // 最多重试3次，每次间隔2秒
//                        int retries = 3;
//                        while (retries-- > 0) {
//                            try {
//                                Object o = operations.opsForList().rightPop("seckill:queue", 30, TimeUnit.SECONDS);
//                                return o;
//                            } catch (RedisCommandTimeoutException e) {
//                                System.err.println("Redis操作超时，剩余重试次数: " + retries);
//                                try { Thread.sleep(2000); } catch (InterruptedException ie) { /* 处理中断 */ }
//                            }
//                        }
//                        throw new RedisCommandTimeoutException("超过最大重试次数");
//                    }
//                });
                // 添加空值检查
                if (task == null) {
                    System.out.println("队列空闲等待中...");
                    continue;
                }
                //取出Redis队列任务
//                Object o = redisTemplate.opsForList().rightPop("seckill:queue", 30, TimeUnit.SECONDS);

                // 阻塞式获取队列任务（30秒超时）
//                Object task = redisTemplate.opsForList().rightPop("seckill:queue", 30, TimeUnit.SECONDS);
                // 添加任务接收日志
                if (task != null) {
                    System.out.println("成功接收任务: " + task);
                } else {
                    System.out.println("等待超时，未获取到任务");
                    continue;
                }
                // 任务类型校验
                if (task instanceof Map) {
                    Map<?, ?> taskMap = (Map<?, ?>) task;
                    String userId = (String) taskMap.get("userId");
                    String productId = (String) taskMap.get("productId");

                    // 执行秒杀核心业务
                    orderService.insert(userId, productId);
                    //修改订单状态
                    seckillService.updateOrderStatus(userId, productId, "2");
                    //删除队列任务
                    redisTemplate.opsForList().remove("seckill:queue", 1, task);
                }else {
                    System.err.println("收到非预期数据类型: " + task.getClass().getName());
                }
            } catch (Exception e) {
                consumerHealthIndicator.setConsumerRunning(false);
                System.err.println("任务处理异常: " + e.getMessage());

                if (e instanceof RedisCommandTimeoutException) {
                    System.err.println("严重：连续多次Redis超时，请检查Redis服务器状态");
                }
                // 异常处理（记录日志 + 告警通知）
                e.printStackTrace();
            }
        }
    }
    @PreDestroy
    public void shutdown() {
        executor.shutdownNow();
    }

}
